package com.demo.spark.rdd;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.List;
import java.util.Random;

public class RddAggregateByKey {
    public static void main(String[] args) {

        SparkConf sparkConf = new SparkConf();
        sparkConf.setAppName("rdd count by key");
        sparkConf.setMaster("local[2]");
        //sparkConf.setMaster("yarn-client");
        JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);
        javaSparkContext.setLogLevel("ERROR");


        List<Integer> data = Arrays.asList(5, 1, 1, 4, 4, 2, 2,5, 1, 1, 4, 4, 2, 21, 1, 4, 4, 2, 2,5, 1, 1, 4, 4, 2, 21, 1, 4, 4, 2, 2,5, 1, 1, 4, 4, 2, 2);
        int numPartitions = 100;
        JavaRDD<Integer> javaRDD = javaSparkContext.parallelize(data);
        final Random random = new Random(100);
        JavaPairRDD<Integer,Integer> javaPairRDD = javaRDD.mapToPair(new PairFunction<Integer, Integer, Integer>() {
            @Override
            public Tuple2<Integer, Integer> call(Integer integer) throws Exception {
                return new Tuple2<Integer, Integer>(integer,random.nextInt(10));
            }
        });

        JavaPairRDD<Integer, Integer> aggregateByKeyRDD = javaPairRDD.aggregateByKey(0,numPartitions, new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return Math.max(v1, v2);
            }
        }, new Function2<Integer, Integer, Integer>() {
            int i = 0;
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return Math.max(v1, v2);
            }
        });
        System.out.println("aggregateByKeyRDD.partitions().size()~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~"+aggregateByKeyRDD.partitions().size());
        aggregateByKeyRDD.foreach(new VoidFunction<Tuple2<Integer, Integer>>() {
            @Override
            public void call(Tuple2<Integer, Integer> integerIntegerTuple2) throws Exception {
                System.out.println(integerIntegerTuple2._1+","+integerIntegerTuple2._2);
            }
        });

    }
}
